package kafkaclient;

import com.google.gson.JsonObject;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class KafkaProducer {


  private static final DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
  private static final String bootstrapServers = "127.0.0.1:9092";

  public static void main(String[] args) {

    int THREADS = 2;
    ExecutorService service = Executors.newFixedThreadPool(THREADS);
    service.submit(new Runnable() {
      @Override
      public void run() {
        task("test1", 2);
      }
    });
    service.submit(new Runnable() {
      @Override
      public void run() {
        task("test2", 1000);
      }
    });

    service.shutdown();

  }

  public static void task(String topic, int nums){

    Properties props = new Properties();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

    Producer<String, String> producer = new org.apache.kafka.clients.producer.KafkaProducer<>(props);


    try {

      for (int i = 0; i < nums; i++) {
        //String format = now.plusSeconds(30L * i).format(formatter);
        LocalDateTime now = LocalDateTime.now();
        String format = now.format(formatter);
        JsonObject jsonObject = new JsonObject();
        jsonObject.addProperty("stime", format);
        jsonObject.addProperty("name", "aa");
        jsonObject.addProperty("val", i);
        String msg = jsonObject.toString();
        ProducerRecord<String, String> record = new ProducerRecord<>(topic, "key", msg);
        producer.send(record);
        Thread.sleep(10000);
      }
    } catch (Throwable e) {
      e.printStackTrace();
    } finally {
      producer.close();
    }
  }
}